package publishSubscribe;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import utils.RabbitMQUtils;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

/**
 * @author： pang-yun
 * @date： 2021-07-30 14:41
 * 功能目的 ：用于  比较 三种发布策略的 快慢
 */

public class ConfirmMessage {
    public static final int QUEUE_COUNT = 1000;

    public static void main(String[] args) throws InterruptedException, TimeoutException, IOException {
     /*1.单个确认
        2.批量确认
        3. 异步批量确认
     * */

//        publishMessageIndividually();     //975ms
//        publicMessageBatch();               // 117 ms
        publishMessageAsync();              // 99ms
    }

    //单个确认  安全但是耗费时间较长
    public static void publishMessageIndividually() throws IOException, TimeoutException, InterruptedException {
        Channel channel = RabbitMQUtils.getChannel();
        //声明队列
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, false, false, false, null);

        //开启发布确认
        channel.confirmSelect();
        //开始时间
        long begin = System.currentTimeMillis();

        //批量发消息
        for (int i = 0; i < QUEUE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes(StandardCharsets.UTF_8));

            boolean confirms = channel.waitForConfirms();
            if (confirms) {
                System.out.println("发送成功 " + i);
            }
        }
        long end = System.currentTimeMillis();
        System.out.println(end - begin);

    }

    // 批量发送消息    速度快 但是无法确定那个消息未被确认
    public static void publicMessageBatch() throws IOException, TimeoutException, InterruptedException {
        Channel channel = RabbitMQUtils.getChannel();
        //声明队列
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, false, false, false, null);

        //开启发布确认
        channel.confirmSelect();
        //开始时间
        long begin = System.currentTimeMillis();

        //批量发消息
        for (int i = 0; i < QUEUE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes(StandardCharsets.UTF_8));

            if (i % 100 == 0) {
                boolean confirms = channel.waitForConfirms();
                if (confirms) {
                    System.out.println("发送成功 " + i);
                }
            }
        }
        long end = System.currentTimeMillis();
        System.out.println(end - begin);
    }

    // 最优策略  异步发送代码
    public static void publishMessageAsync() throws IOException, TimeoutException {
        Channel channel = RabbitMQUtils.getChannel();
        //声明队列
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, false, false, false, null);

        //开启发布确认
        channel.confirmSelect();

        /**
         * 线程安全有序的一个哈希表， 适用于高并发的环境下
         * 1.将消息与序号关联
         * 2。轻松批量删除 条目
         * 3.支持高并发 多线程
         */
        ConcurrentSkipListMap<Long, String> map = new ConcurrentSkipListMap<>();


        //消息发送成功回调函数
        ConfirmCallback ack = (deliveryTag, multiple) -> {
            // 2。从记录中删除已经发送成功的消息
            // 批量与 不是批量删除
            if (multiple) {
                ConcurrentNavigableMap<Long, String> navigableMap = map.headMap(deliveryTag);
                navigableMap.clear();
            } else {
                map.remove(deliveryTag);
            }

            System.out.println("确认消息编号 " + deliveryTag);
        };
        //消息失败回调函数
        ConfirmCallback failAck = (deliveryTag, multiple) -> {
            // 3. 打印 一下为确认的消息都有哪些？
            System.out.println("未确认信息 " + deliveryTag);
        };

        /**
         * 监听那些消息成功  监听那些消息失败   异步
         */
        channel.addConfirmListener(ack, failAck);

        //开始时间
        long begin = System.currentTimeMillis();
        for (int i = 0; i < QUEUE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes(StandardCharsets.UTF_8));


            // 1.记录所有需要发送的消息
            map.put(channel.getNextPublishSeqNo(), message);
        }
        long end = System.currentTimeMillis();
        System.out.println(end - begin);
    }


}
